package com.szwistar.common.comm.jsonrpc.mina;

import com.shephertz.appwarp.websocket.binary.WebSocketCodecFactory;
import com.szwistar.common.ERR;
import com.szwistar.common.comm.CommEndpoint;
import com.szwistar.common.comm.P2pBase;
import com.szwistar.common.comm.jsonrpc.*;
import com.szwistar.common.datastruct.JsonMapper;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.LineDelimiter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.net.URI;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

/**
 * 连接端点
 * <li>是 MinaJsonRpcServer 服务器 accept 的一个客户端连接
 * <li>或是 P2pBase 主动连接到 MinaJsonRpcServer 服务器
 * @author 张勇
 */
public class MinaJsonRpcCommEndpoint extends JsonRpcEndpointAbstract implements CommEndpoint {
    protected static final Logger LOG = LoggerFactory.getLogger(MinaJsonRpcCommEndpoint.class);

    // 连接实体
    protected IoSession session;
    // 根据客户端地址生成客户端唯一标识
    protected String from;
    // 对端订阅的主题列表
    protected List<String> peerSubscribedTopics = new ArrayList<String>();

    @Override
    public String toString() {
        return "MinaJsonRpcCommEndpoint@" + from;
    }

    public String getFrom() { return from; }
    @Override
    public IoSession getSession() { return session; }
    @Override
    public SocketAddress getLocalAddress() { return (session == null) ? null : session.getLocalAddress(); }
    @Override
    public SocketAddress getPeerAddress() { return (session == null) ? null : session.getRemoteAddress(); }
    @Override
    public boolean isConnected() { return session != null && session.isConnected(); }

    /**
     * 构造函数应设置为 public，但仅供 MinaJsonRpcServer accept() 时自动创建实例使用。
     * @apiNote 要创建 P2P 实例，应调用 connect() 方法创建！
     */
    public MinaJsonRpcCommEndpoint() {}

    @Override
    public void close() {
        if(session == null) { return; }
        if(session.isConnected()) {	session.close(true); }
        session = null;
    }

    @Override
    public WriteFuture write(Object message) {
        if(!isConnected()) { return null; }
        return session.write(message);
    }

    @Override
    public void onCreate(IoSession session, Object context) throws Exception {
        // 保存连接
        this.session = session;
        // 在连接建立时，根据客户端地址生成客户端唯一标识
        from = session.getRemoteAddress().toString();
        // 初始化
        if(context instanceof MinaJsonRpcServer) {
            // 如果是 Server 创建的 PeerSocket 则 server 有效
            // 如果是 P2pBase 连接，则 server 无效
            MinaJsonRpcServer server = (MinaJsonRpcServer) context;
            // 替换基类 JsonRpcEndpointAbstract 中的相关变量
            myLocalTopic = server.myLocalTopic;
            methodsMap = server.methodsMap;
            eventsMap = server.methodsMap;
        }
    }

    @Override
    public void onOpened() {}

    @Override
    public void onMessageSent(Object message) {}

    @Override
    public void onMessageReceived(Object message) {
        JsonRpcMessage msg = null;
        String json = null;
        try {
            // 消息解码
            json = message.toString();
            msg = JsonMapper.fromJson(json, JsonRpcMessage.class);
        } catch (Exception e) {
            LOG.error("RPC 接收消息解码失败！\n{}\n{}", json, e.getMessage());
            return;
        }

        try {
            if(msg instanceof JsonRpcMessageRequest) {
                // !!!注意：一定要替换消息中的客户端标识!!!
                ((JsonRpcMessageRequest)msg).setFrom(from);
            }

            // 标记消息的接收接口（如果已经标记了的话，就不要覆盖它了，可能是派生类标记过了）
            msg.setEndpoint(this);

            // 调用基类接收消息处理函数
            onMessageReceived(msg);
        } catch (Exception e) {
            LOG.error("RPC 接收消息处理失败！", e);
        }
    }

    @Override
    public void onClosed() {
        LOG.debug("{} closed.", this);
        session = null;
    }

    @Override
    public void onIdle(IdleStatus status) {}

    @Override
    public void onExceptionCaught(Throwable cause) {}

    @Override
    public void listenOn(Object linkObj, String caller) throws Exception {}

    @Override
    public void send(String target, JsonRpcMessage msg, int qos) throws Exception {
        String jsonMsg = JsonMapper.toJson(msg);

        if(msg instanceof JsonRpcMessageNotify) {
            // 如果客户端没有监听此消息，就不用发送了?
            //if(!isPeerSubscribed(target)) return;
            LOGCOMM.debug("向 '{}' 发送主题为 '{}' 的通知：{}", this, target, msg);
            write(jsonMsg);
        } else {
            write(jsonMsg);
        }
    }

    /**
     * 判断某个对端是否订阅了某个主题的事件
     * @param topic 主题
     * @return 是否已经预订
     */
    protected boolean isPeerSubscribed(String topic) {
        return JsonRpcEndpointAbstract.filterTopic(peerSubscribedTopics, topic);
    }

    @Override
    public JsonRpcMessageReturn handleSubscribe(JsonRpcMessageCall request) {
        Object[] filters;

        // 要添加的主题
        filters = request.getParaToArray("add", new String[0]);
        for(Object item: filters) {
            String filter = (String) item;
            if(!JsonRpcEndpointAbstract.validateFilter(filter)) { return JsonRpcMessageReturn.error(request, ERR.BAD_PARAM, "错误的主题过滤器格式！"); }
            if(!peerSubscribedTopics.contains(filter)) { peerSubscribedTopics.add(filter); }
        }

        // 要删除的主题
        filters = request.getParaToArray("del", new String[0]);
        for(Object item: filters) {
            String filter = (String) item;
            peerSubscribedTopics.remove(filter);
        }

        return JsonRpcMessageReturn.ok(request, null);
    }

    /**
     * 连接到 JsonRpc@TCP/UDP 服务器
     * <p>数据以 JSON 格式字符串传输，以 \n 做为结束符；中文使用 GBK 编码。
     * @param host 服务器监听地址或域名
     * @param port 服务器监听端口
     * @param tcp true: 使用 TCP，false: 使用 UDP
     */
    public static MinaJsonRpcCommEndpoint connectSocket(String host, int port, boolean tcp) throws Exception {
        // 设置编解码器
        String encode = "GBK";
        String decodingDelimiter = LineDelimiter.UNIX.getValue();
        String encodingDelimiter = LineDelimiter.UNIX.getValue();

        TextLineCodecFactory codecFactory = new TextLineCodecFactory(Charset.forName(encode), encodingDelimiter, decodingDelimiter);
        codecFactory.setDecoderMaxLineLength(256*1024);
        codecFactory.setEncoderMaxLineLength(256*1024);

        P2pBase<MinaJsonRpcCommEndpoint> client = new P2pBase<>(host, port, MinaJsonRpcCommEndpoint.class, new ProtocolCodecFilter(codecFactory), tcp);
        return client.open(null);
    }

    /**
     * 连接到 JsonRpc@WebSocket 服务器
     * @param host 服务器监听地址或域名
     * @param port 服务器监听端口
     */
    public static MinaJsonRpcCommEndpoint connectWebSocket(String host, int port) throws Exception {
        P2pBase<MinaJsonRpcCommEndpoint> client = new P2pBase<>(host, port, MinaJsonRpcCommEndpoint.class, new ProtocolCodecFilter(new WebSocketCodecFactory()), true);
        return client.open(null);
    }

    /**
     * 根据地址标识，确定连接类型，并连接到 MinaJsonRpc 服务器
     * <p>支持的地址格式如：
     *  <li>JsonRpc@WebSocket服务器地址，如 ws://192.168.123.4:8504/wskt
     *  <li>JsonRpc@WebSocketSsl 服务器地址，如 wss://192.168.123.4:8504/wskt
     *  <li>JsonRpc@TcpSocket服务器地址，如 tcp://192.168.123.4:8503
     *  <li>JsonRpc@UdpSocket服务器地址，如 udp://192.168.123.4:8502
     * @param address 服务器地址。
     */
    public static MinaJsonRpcCommEndpoint connect(String address) throws Exception {
        URI uri = new URI(address);

        // 根据地址，确定连接类型
        if(uri.getScheme().equals("ws") || uri.getScheme().equals("wss")) {
            return connectWebSocket(uri.getHost(), uri.getPort());
        } else if(uri.getScheme().equals("tcp")) {
            return connectSocket(uri.getHost(), uri.getPort(), true);
        } else if(uri.getScheme().equals("udp")) {
            return connectSocket(uri.getHost(), uri.getPort(), false);
        } else {
            throw new Exception("不支持的地址：" + address);
        }
    }
}
